package com.atguigu.edu.realtime.dwd.dbsplit.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.HashMap;
import java.util.Map;

import static jdk.nashorn.internal.runtime.regexp.joni.Config.log;

public class DwdBaseLog extends BaseApp {
    private final String ERR = "err";
    private final String START = "start";
    private final String DISPLAY = "display";
    private final String ACTION = "action";
    private final String PAGE = "page";
    private  final String APPVIDEO = "appvideo";
    public static void main(String[] args) {
        new DwdBaseLog().start(
                10011,
                4,
                "dwd_base_log",
                Constant.TOPIC_LOG
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {

        //简单etl
        SingleOutputStreamOperator<JSONObject> jsonObjDs = kafkaStrDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    return jsonObject;
                } catch (Exception e) {
                    log.print("日志格式不是正确的 JSON 格式: ");
                }
                return null;
            }
        });
        //修复新老访客
        KeyedStream<JSONObject, String> keyedDs = jsonObjDs.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                String mid = jsonObject.getJSONObject("common").getString("mid");
                return mid;
            }
        });
        SingleOutputStreamOperator<JSONObject> fixedDs = keyedDs.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            ValueState<String> lastVisitDateState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<String>("lastVistDateState", String.class);
                lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                String isNew = jsonObject.getJSONObject("common").getString("is_new");
                //从状态中获取上次访问日期
                String lastVisitDate = lastVisitDateState.value();
                //获取当前访问日期
                Long ts = jsonObject.getLong("ts");
                String curVisitDate = DateFormatUtil.tsToDate(ts);

                if ("1".equals(isNew)) {
                    //如果is_new的值为1
                    if (StringUtils.isEmpty(lastVisitDate)) {
                        //如果键控状态为null，认为本次是该访客首次访问 APP，将日志中 ts 对应的日期更新到状态中，不对 is_new 字段做修改；
                        lastVisitDateState.update(curVisitDate);
                    } else {
                        //如果键控状态不为null，且首次访问日期不是当日，说明访问的是老访客，将 is_new 字段置为 0；
                        if (!curVisitDate.equals(lastVisitDate)) {
                            isNew = "0";
                            jsonObject.getJSONObject("common").put("is_new", isNew);
                        }
                    }
                } else {
                    //如果 is_new 的值为 0
                    //如果键控状态为 null，说明访问 APP 的是老访客但本次是该访客的页面日志首次进入程序。当前端新老访客状态标记丢失时，
                    // 日志进入程序被判定为新访客，Flink 程序就可以纠正被误判的访客状态标记，只要将状态中的日期设置为今天之前即可。
                    // 本程序选择将状态更新为昨日；
                    if (StringUtils.isEmpty(lastVisitDate)) {
                        long yesterdayTs = ts - 24 * 60 * 60 * 1000;
                        String yesterDay = DateFormatUtil.tsToDate(yesterdayTs);
                        lastVisitDateState.update(yesterDay);
                    }
                }
                collector.collect(jsonObject);
            }
        });
        //分流 将不同类型的日志放到不同的流中
        OutputTag<String> errTag = new OutputTag<String>("errTag"){};
        OutputTag<String> startTag = new OutputTag<String>("startTag"){};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag"){};
        OutputTag<String> actionTag = new OutputTag<String>("actionTag"){};
        OutputTag<String> appVideoTag = new OutputTag<String>("appVideoTag"){};
        SingleOutputStreamOperator<String> pageDs = fixedDs.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                //错误日志
                JSONObject errJsonObj = jsonObject.getJSONObject("err");
                if (errJsonObj != null) {
                    context.output(errTag, jsonObject.toJSONString());
                    jsonObject.remove("err");
                }
                //appVideo
                JSONObject appVideo = jsonObject.getJSONObject("appVideo");

                if (appVideo != null) {
                    JSONObject newAppVideojsonobj = new JSONObject();
                    JSONObject commonJsonObj = jsonObject.getJSONObject("common");
                    Long ts = jsonObject.getLong("ts");
                    newAppVideojsonobj.put("appVideo",appVideo);
                    newAppVideojsonobj.put("common",commonJsonObj);
                    newAppVideojsonobj.put("ts",ts);
                    context.output(appVideoTag,newAppVideojsonobj.toJSONString());
                }

                //启动日志
                JSONObject stratJsonobj = jsonObject.getJSONObject("strat");
                if (stratJsonobj != null) {
                    context.output(startTag, jsonObject.toJSONString());
                } else {
                    JSONObject commonJsonObj = jsonObject.getJSONObject("common");
                    JSONObject pageJsonObj = jsonObject.getJSONObject("page");
                    Long ts = jsonObject.getLong("ts");
                    //display
                    JSONArray displayJsonArr = jsonObject.getJSONArray("displays");
                    if (displayJsonArr != null && displayJsonArr.size() > 0) {
                        for (int i = 0; i < displayJsonArr.size(); i++) {
                            JSONObject displayJsonObj = displayJsonArr.getJSONObject(i);
                            JSONObject newDisplayJsonObj = new JSONObject();
                            newDisplayJsonObj.put("common", commonJsonObj);
                            newDisplayJsonObj.put("page", pageJsonObj);
                            newDisplayJsonObj.put("display", displayJsonObj);
                            newDisplayJsonObj.put("ts", ts);
                            context.output(displayTag, newDisplayJsonObj.toJSONString());
                        }
                        jsonObject.remove("displays");
                        //actions
                        JSONArray actions = jsonObject.getJSONArray("actions");
                        if (actions != null && actions.size() > 0) {
                            for (int i = 0; i < actions.size(); i++) {
                                JSONObject actionJsonObj = actions.getJSONObject(i);
                                JSONObject newActionjsonobj = new JSONObject();
                                newActionjsonobj.put("common", commonJsonObj);
                                newActionjsonobj.put("page", pageJsonObj);
                                newActionjsonobj.put("action", newActionjsonobj);
                                context.output(actionTag, newActionjsonobj.toJSONString());
                            }
                            jsonObject.remove("actions");
                        }
                        collector.collect(jsonObject.toJSONString());
                    }

                }
            }
        });
        //pageDs.print();
        SideOutputDataStream<String> errDs = pageDs.getSideOutput(errTag);
        SideOutputDataStream<String> startDs = pageDs.getSideOutput(startTag);
        SideOutputDataStream<String> displayDs = pageDs.getSideOutput(displayTag);
        SideOutputDataStream<String> actionDs = pageDs.getSideOutput(actionTag);
        SideOutputDataStream<String> appVideoDs = pageDs.getSideOutput(appVideoTag);
        Map<String, DataStream> dsMap = new HashMap<>();
        dsMap.put(ERR,errDs);
        dsMap.put(START,startDs);
        dsMap.put(DISPLAY,displayDs);
        dsMap.put(ACTION,actionDs);
        dsMap.put(PAGE,pageDs);
        dsMap.put(APPVIDEO,appVideoDs);

        dsMap
                .get(PAGE)
                .sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PAGE));
        dsMap
                .get(ERR)
                .sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));
        dsMap
                .get(START)
                .sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_START));
        dsMap
                .get(DISPLAY)
                .sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_DISPLAY));
        dsMap
                .get(ACTION)
                .sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ACTION));
        dsMap
                .get(APPVIDEO)
                .sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_APP_VIDEO));

    }
}
